package cn.baker.delayqueue.init;


import cn.baker.delayqueue.config.DelayQueueProperties;
import cn.baker.delayqueue.util.RedisDelayQueueUtil;
import cn.baker.delayqueue.service.RedisDelayQueueHandle;
import com.alibaba.fastjson.JSON;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.base.Throwables;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.CommandLineRunner;

import java.util.List;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * 启动延迟队列监测扫描
 *
 * @author yuanyu
 */
@Slf4j
public class RedisDelayQueueRunner implements CommandLineRunner {

    private final RedisDelayQueueUtil redisDelayQueueUtil;
    private final List<RedisDelayQueueHandle> redisDelayQueueHandleList;
    private final ObjectMapper objectMapper;
    private final DelayQueueProperties delayQueueProperties;

    public RedisDelayQueueRunner(RedisDelayQueueUtil redisDelayQueueUtil, List<RedisDelayQueueHandle> redisDelayQueueHandleList, ObjectMapper objectMapper, DelayQueueProperties delayQueueProperties) {
        this.redisDelayQueueUtil = redisDelayQueueUtil;
        this.redisDelayQueueHandleList = redisDelayQueueHandleList;
        this.objectMapper = objectMapper;
        this.delayQueueProperties = delayQueueProperties;


    }

    private final ThreadPoolExecutor runExecutor = new ThreadPoolExecutor(1, 1, 0, TimeUnit.SECONDS, new LinkedBlockingQueue<>(1), new ThreadFactoryBuilder().setNameFormat("delay-queue-%d").build());
    private final ThreadPoolExecutor scanExecutor = new ThreadPoolExecutor(3, 5, 30, TimeUnit.SECONDS, new LinkedBlockingQueue<>(1000), new ThreadFactoryBuilder().setNameFormat("order-delay-%d").build());

    @Override
    public void run(String... args) {
        runExecutor.execute(() -> {
            while (true) {
                scan();
            }
        });
        log.info("redisson延迟队列监测启动成功");
    }


    private void scan() {
        try {
            for (RedisDelayQueueHandle redisDelayQueueHandle : redisDelayQueueHandleList) {
                String value = redisDelayQueueUtil.getDelayQueue(redisDelayQueueHandle.queueCode());
                if (value != null) {
                    scanExecutor.execute(() -> {
                        Object obj;
                        try {
                            obj = JSON.parseObject(value, redisDelayQueueHandle.getDtoClass());
                            redisDelayQueueHandle.execute(obj);
                        } catch (Exception e) {
                            log.error("e={}", Throwables.getStackTraceAsString(e));
                        }
                    });
                }
            }
            TimeUnit.MILLISECONDS.sleep(delayQueueProperties.getQueueScanFreq());
        } catch (InterruptedException e) {
            log.error("redisson延迟队列监测异常中断 {}", e.getMessage());
        }
    }
}
